package com.niit.MostPopularMovieDecadeStart1980;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;

public class MPMDSReducer extends Reducer<Text, MPMDSBean, Text, MPMDSBean>{
    // 定义一个TreeMap作为存储数据的容器（按key排序）
    // 80+ 90+ 00+ 10+ 20+
    private TreeMap<MPMDSBean, Text> flowMap80 = new TreeMap<>();
    private TreeMap<MPMDSBean, Text> flowMap90 = new TreeMap<>();
    private TreeMap<MPMDSBean, Text> flowMap00 = new TreeMap<>();
    private TreeMap<MPMDSBean, Text> flowMap10 = new TreeMap<>();
    @Override
    protected void reduce(Text key, Iterable<MPMDSBean> values, Context context) throws IOException, InterruptedException {
        System.out.println(values);
        for (MPMDSBean value : values) {
            System.out.println(value);
            MPMDSBean flowBean = new MPMDSBean();
            flowBean.set(value.getRate(),value.getRank(),value.getTypes(),value.getRegions(),value.getTitle()
            ,value.getRelease_data(),value.getActor_count(),value.getVote_count(),value.getScore(),value.getActors());

            Text movie = new Text(key);


//             1 向treeMap集合中添加数据

            if(value.getRelease_data().charAt(2) == '8') flowMap80.put(flowBean,movie);
            if(value.getRelease_data().charAt(2) == '9') flowMap90.put(flowBean,movie);
            if(value.getRelease_data().charAt(2) == '0' && value.getRelease_data().charAt(0) == '2') flowMap00.put(flowBean,movie);
            if(value.getRelease_data().charAt(2) == '1' && value.getRelease_data().charAt(0) == '2') flowMap10.put(flowBean,movie);


            // 2 限制TreeMap数据量，超过10条就删除掉流量最小的一条数据
            if (flowMap80.size() > 10) {
                flowMap80.remove(flowMap80.lastKey());
            }
            if (flowMap90.size() > 10) {
                flowMap90.remove(flowMap90.lastKey());
            }
            if (flowMap00.size() > 10) {
                flowMap00.remove(flowMap00.lastKey());
            }
            if (flowMap10.size() > 10) {
                flowMap10.remove(flowMap10.lastKey());
            }
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // 3 遍历集合，输出数据
        Iterator<MPMDSBean> iterator80 = flowMap80.keySet().iterator();
        Iterator<MPMDSBean> iterator90 = flowMap90.keySet().iterator();
        Iterator<MPMDSBean> iterator00 = flowMap00.keySet().iterator();
        Iterator<MPMDSBean> iterator10 = flowMap10.keySet().iterator();

        while (iterator80.hasNext()) {
            MPMDSBean valueOut = iterator80.next();
            Text keyOut = flowMap80.get(valueOut);
            context.write(keyOut, valueOut);
        }
        while (iterator90.hasNext()) {
            MPMDSBean valueOut = iterator90.next();
            Text keyOut = flowMap90.get(valueOut);
            context.write(keyOut, valueOut);
        }
        while (iterator00.hasNext()) {
            MPMDSBean valueOut = iterator00.next();
            Text keyOut = flowMap00.get(valueOut);
            context.write(keyOut, valueOut);
        }
        while (iterator10.hasNext()) {
            MPMDSBean valueOut = iterator10.next();
            Text keyOut = flowMap10.get(valueOut);
            context.write(keyOut, valueOut);
        }

    }
}
